package com.cryptoneedle.garden.doris.fix.task;

import com.cryptoneedle.garden.entity.garden.doris.DorisTable;
import com.cryptoneedle.garden.entity.garden.doris.DorisTableScript;
import com.cryptoneedle.garden.enums.DorisTableModel;
import com.cryptoneedle.garden.repository.doris.DorisMetadataRepository;
import com.cryptoneedle.garden.service.crud.garden.doris.*;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Strings;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * <p>description:  </p>
 *
 * @author CryptoNeedle
 * @date 2025-11-19
 */
@Slf4j
@Component
public class DorisBucketFixTask implements ApplicationRunner {

    @Autowired
    private AddDorisService add;
    @Autowired
    private DeleteDorisService delete;
    @Autowired
    private FoundDorisService select;
    @Autowired
    private PatchDorisService patch;
    @Autowired
    private SaveDorisService save;

    @Resource
    private JdbcTemplate dorisJdbcTemplate;
    @Autowired
    private DorisMetadataRepository dorisMetadataRepository;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        Thread.sleep(3 * 1000);

        // todo 从配置中提取
        String[] databases = new String[]{"seatone_ads", "seatone_dim", "seatone_dwd", "seatone_dwd_standard", "seatone_dws", "seatone_mapping", "seatone_mapping_standard", "seatone_ods", "seatone_pre"};
        List<DorisTable> dorisTables = select.dorisTables()
                .stream()
                .filter(table -> Strings.CI.containsAny(table.getDatabaseName(), databases))
                .toList();
        int tableMaxLength = dorisTables.stream()
                .map(table -> table.getDatabaseName() + "." + table.getTableName())
                .mapToInt(String::length)
                .max()
                .orElse(0);

        int tableNum = dorisTables.size();
        for (int i = 0; i < tableNum; i++) {
            DorisTable dorisTable = dorisTables.get(i);
            String tableName = dorisTable.getTableName();
            String databaseName = dorisTable.getDatabaseName();
            String bucketNum = dorisTable.getBucket();
            Integer estimateBucket = dorisTable.getEstimateBucket();
            DorisTableModel tableModel = dorisTable.getTableModel();

            DorisTableScript createTable = dorisMetadataRepository.showCreateTable(dorisTable.getCatalogName(), databaseName, tableName);
            String createTableSql = createTable.getScript();

            String comment = ("-- [%-4s/%-4s] %-" + tableMaxLength + "s | Storage(include 3 copies): %-9s Mb | ReplicaCount: %-3s | Bucket: %-4s | EstimateBucket: %-2s")
                    .formatted(i + 1, tableNum, databaseName + "." + tableName, dorisTable.getSizeFormatMegaBytes(), dorisTable.getReplicaCount(), bucketNum, estimateBucket);
            log.info(comment);

            // 不进行处理的情况
            if (DorisTableModel.AGGREGATE_KEY.equals(tableModel)) {
                log.warn("Skip AGGREGATE KEY Table: {}", tableName);
                continue;
            }
            if (DorisTableModel.DUPLICATE_KEY.equals(tableModel)) {
                log.warn("Skip DUPLICATE KEY Table: {}", tableName);
                continue;
            }
            if (dorisTable.getPartition()) {
                log.warn("Skip PARTITION BY Table: {}", tableName);
                continue;
            }
            if (estimateBucket == null) {
                log.warn("Skip EstimateBucket NULL Table: {}", tableName);
                continue;
            }

            String bucket = estimateBucket.toString();
            if (Strings.CI.equals(bucketNum, bucket)) {
                //log.warn("Skip Equal EstimateBucket Table: {}", tableName);
                continue;
            }

            // 临时表名 fix_xxx
            String newTableName = "fix_" + tableName;
            String newCreateTableSql = Strings.CS.replace(createTableSql, "CREATE TABLE", "CREATE TABLE IF NOT EXISTS");
            newCreateTableSql = Strings.CS.replace(newCreateTableSql, "`%s`".formatted(tableName), "`%s`.`%s`".formatted(databaseName, newTableName));
            newCreateTableSql = StringUtils.substringBefore(newCreateTableSql, "BUCKETS");
            newCreateTableSql = newCreateTableSql
                    + "BUCKETS " + estimateBucket
                    + """
                     \nPROPERTIES (
                        "replication_num" = "3",
                        "is_being_synced" = "false",
                        "compression" = "LZ4",
                        "enable_unique_key_merge_on_write" = "true",
                        "light_schema_change" = "true",
                        "enable_mow_light_delete" = "false",
                        "store_row_column" = "true"
                    );
                    """;

            String replaceTableData = "INSERT INTO `%s`.`%s` SELECT * FROM `%s`.`%s`;".formatted(databaseName, newTableName, databaseName, tableName);
            String dropTable = "DROP TABLE IF EXISTS `%s`.`%s`;".formatted(databaseName, tableName);
            String renameTable = "ALTER TABLE `%s`.`%s` RENAME %s;".formatted(databaseName, newTableName, tableName);

            if (false) {
                // todo 保存记录
                log.info(createTableSql);
                log.info(newCreateTableSql);
                log.info(replaceTableData);
                log.info(dropTable);
                log.info(renameTable);
            }

            // 新建表语句
            dorisJdbcTemplate.execute(newCreateTableSql);
            // 数据替换
            dorisJdbcTemplate.execute(replaceTableData);
            // 删除旧表
            dorisJdbcTemplate.execute(dropTable);
            // 重命名新表
            dorisJdbcTemplate.execute(renameTable);
        }
    }
}